簡介

上一篇講解系統開發的流程，今天會介紹 pyramid 開發、測試，還有常用的測試工具，

models

mymodel.py : 定義 ORM 資料表

class MyModel(Base):
    __tablename__ = 'models'
    id = Column(Integer, primary_key=True)
    name = Column(Text)
    value = Column(Integer)
Index('my_index', MyModel.name, unique=True, mysql_length=255)

API

routes.py : 定義 Router 路徑

from .views import api

def includeme(config):
    # API
    config.add_route('api', '/api')
    config.add_view(
        api.ApiView, attr='create',
        route_name='api',
        request_method='POST',
        renderer='json'
    )
    config.add_view(
        api.ApiView, attr='search',
        route_name='api',
        request_method='GET',
        renderer='json'
    )

    config.add_route('api.info', '/api/{id}')
    config.add_view(
        api.ApiView, attr='get',
        route_name='api.info',
        request_method='GET',
        renderer='json'
    )

    config.add_view(
        api.ApiView, attr='edit',
        route_name='api.info',
        request_method='POST',
        renderer='json'
    )

    config.add_view(
        api.ApiView, attr='delete',
        route_name='api.info',
        request_method='DELETE',
        renderer='json'
    )

api.py : 主要邏輯

# coding=utf8
from __future__ import unicode_literals
import transaction
from ..models import MyModel

class ApiView(object):
    def __init__(self, request):
        super(ApiView, self).__init__()
        self.request = request
        self.dbsession = self.request.dbsession

    def create(self):
        """
        創建 API
        """
        try:
            name = self.request.params.get('name')
            value = self.request.params.get('value')

            with transaction.manager:
                self.dbsession.add(MyModel(name=name, value=value))

            mymodel = self.dbsession.query(MyModel).\
                           filter(MyModel.name == name).first()

            return {'code': 200, 'status': True,
                    'response': {'id': mymodel.id, 
                                 'name': mymodel.name, 
                                 'value': mymodel.value}}

        except Exception as e:
            return {'code': 500, 'status': False, 'response': str(e)}


    def search(self):
        """
        搜尋 API
        """
        try:
            mymodel_list = self.dbsession.query(MyModel).all()
            mymodel_json_list = []
            for mymodel in mymodel_list:
                mymodel_json_list.append({'id': mymodel.id, 
                                          'name': mymodel.name, 
                                          'value': mymodel.value})

            return {'code': 200, 'status': True,
                    'response': mymodel_json_list}

        except Exception as e:
            return {'code': 501, 'status': False, 'response': str(e)}


    def get(self):
        """
        讀取 API
        """
        try:
            mymodel = self.dbsession.query(MyModel)\
                          .filter(MyModel.id == int(self.request.matchdict['id']))\
                          .first()

            return {'code': 200, 'status': True,
                    'response': {'id': mymodel.id, 
                                 'name': mymodel.name, 
                                 'value': mymodel.value}}

        except Exception as e:
            return {'code': 501, 'status': False, 'response': str(e)}

    def edit(self):
        """
        編輯 API
        """
        try:
            name = self.request.params.get('name')
            value = self.request.params.get('value')

            mymodel = self.dbsession.query(MyModel)\
                          .filter(MyModel.id == int(self.request.matchdict['id']))\
                          .first()

            with transaction.manager:
                mymodel.name = name
                mymodel.value = value
                self.dbsession.flush()

            return {'code': 200, 'status': True,
                    'response': {'id': mymodel.id, 
                                 'name': mymodel.name, 
                                 'value': mymodel.value}}

        except Exception as e:
            return {'code': 500, 'status': False, 'response': str(e)}

    def delete(self):
        """
        刪除 API
        """
        try:
            mymodel = self.dbsession.query(MyModel)\
                          .filter(MyModel.id == int(self.request.matchdict['id']))\
                          .first()
            with transaction.manager:
                self.dbsession.delete(mymodel)
                self.dbsession.flush()

            return {'code': 200, 'status': True,
                    'response': 'OK'}

        except Exception as e:
            return {'code': 501, 'status': False, 'response': str(e)}

Postman

創建 API

搜尋 API

讀取 API

修改 API

刪除 API

結論

在上面實做的過程中，因為時間的關係，只是寫一個非常基礎的 CRUD 功能，還缺少一些東西像是，檢查機制、Log機制等，在後續的章節之中會視情況補齊．
有一運用到一些開發的技巧：

1. RestFull API 使用 http Method 來定義相對應的 CRUD 動作
2. 錯誤的處理
3. 使用 Postman 來快速驗證功能是否正常

參考

bitbucket Repostory